package org.myorg;

import java.io.IOException;
import java.util.*;

import java.awt.Color;
import java.awt.image.BufferedImage;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import javax.imageio.ImageIO;
import java.io.InputStream;
import java.io.FileInputStream;
import javax.imageio.stream.*;
import javax.imageio.*;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.*;
import org.apache.hadoop.mapreduce.lib.output.*;
import org.apache.hadoop.util.*;
import org.apache.hadoop.fs.*;

public class AutoProduce extends Configured implements Tool {

	public static class Map extends
			Mapper<NullWritable, BytesWritable, IntWritable, Text> {
		private IntWritable newKey = new IntWritable();
		//private Text result = new Text();
		

		public void map(NullWritable key, BytesWritable value, Context context)
				throws IOException, InterruptedException {
			// If there are 5 frames in each diretory, actually in motion list,
			// there are only 4 items
			// motion[0] = frame_color[1] - frame_color[0];
			// motion[3] = frame_color[4] - frame_color[3];
			/*
			 * 1.get the frame number, such as frame0280.jpeg-----> number =
			 * 0280 2.read its previous frame0279.jpeg 3.compute the
			 * motionEstimate 4.you have to attach the number of cam to the
			 * motionEstimate, ---->Obj:(cam_num, motionEstimate) 5.key = 0279,
			 * context.write(key, Obj)
			 */
			// /data/ap_til/cam1/frame2420.jpeg
			// frame0014.jpeg
			// hdfs://rocks.local:54310/data/ap_example/cam2/frame0014.jpeg
			FileSplit fileSplit = (FileSplit) context.getInputSplit();
			String filename = fileSplit.getPath().getName();
			int fileNum = Integer.parseInt(filename.substring(5,9));
			System.out.println("File name " + filename +" fileNum " + fileNum);
			if(fileNum == 0) {return;}
			int index = fileSplit.getPath().toString().lastIndexOf("/");
			int camNum = Integer.parseInt(fileSplit.getPath().toString().substring(index-4, index).substring(3));
			System.out.println("Directory :" + fileSplit.getPath().toString());
			int number = fileNum - 1; //number = 13
			String str = String.format("%04d",number);
			int lastIndex = fileSplit.getPath().toString().lastIndexOf(".");		
			String preceFile = fileSplit.getPath().toString().replace(fileSplit.getPath().toString().substring(lastIndex-4, lastIndex), str);
			System.out.println("file is " + fileSplit.getPath().toString());
			System.out.println("preceding file is " + preceFile);
			// open cam1/0026.jpeg
			// get motionEstimate (0014.jpeg, 0013.jpeg)			
			BufferedImage frame = null;
			BufferedImage prevFrame = null;
			Path aPath = null, bPath = null;
			FileSystem fs = FileSystem.get(new Configuration());
			aPath = fileSplit.getPath();
			bPath = new Path(preceFile);
			ImageInputStream a = ImageIO.createImageInputStream(fs.open(aPath));
			if(a == null){
				System.out.println("a is null");
				System.out.println("aPath is " + aPath.toString());
			}
			ImageInputStream b = ImageIO.createImageInputStream(fs.open(bPath));
			if(b == null){
				System.out.println("b is null");
				System.out.println("bPath is " + bPath.toString());
			}
			Color fcolor, pfcolor;
			int height = -1;
			int width = -1;
			double motionEstimate = 0;
			ImageInputStream[] frameList = new ImageInputStream[]{b, a};
			for (ImageInputStream f : frameList) {
			
				//sanity check
				//System.out.println("processing file: " + f.getPath());
				
				//if (!f.getPath().endsWith("jpeg") && !f.getPath().endsWith("jpg")) continue;
				
				prevFrame = frame;
				//static BufferedImage	read(ImageInputStream stream) 
				frame = ImageIO.read(f);
				if (prevFrame==null) continue;

				if (height <0) { height = frame.getHeight(); }
				if (width <0) { width = frame.getWidth(); }
				
				int motionSum = 0;
				
				//compare each pixel in this frame to the same pixel in the previous frame
				for (int y=0; y<height; y++) {
					for (int x=0; x<width; x++) {
						
						fcolor = new Color(frame.getRGB(x, y));
						pfcolor = new Color(prevFrame.getRGB(x, y));
						
						motionSum+= Math.abs(fcolor.getRed() - pfcolor.getRed());
						motionSum+= Math.abs(fcolor.getGreen() - pfcolor.getGreen());
						motionSum+= Math.abs(fcolor.getBlue() - pfcolor.getBlue());
						
					}
				}
				
				//A higher number is caused by a significant change in the color values. 
				//This value is interpreted as the amount of motion between sequential frames.
				motionEstimate = motionSum / ((double) height*width);
				//motion.add(motionEstimate);
			}
			// result = "cam1,26,motionEstimate";
			String result = new String(camNum + ","+ Double.toString(motionEstimate));
			//result = camNum + ",";
			//result.append(Double.toString(motionEstimate));
			System.out.println("result is " + result.toString());
			System.out.println("number is " + number);
			newKey.set(number);
			context.write(newKey, new Text(result));
		}
	}

	public static class Reduce extends
			Reducer<IntWritable, Text, IntWritable, IntWritable> {
		public void reduce(IntWritable key, Iterable<Text> values,
				Context context) throws IOException, InterruptedException {
			/*
			 * 1.each key has two items in the list, the key is as same as the
			 * output of the map 2.compare the motionEstimate of them,
			 * if(cam1.motionEstimate > cam2.motionEstimate), choose cam1; else
			 * choose cam2.
			 */
			List<Text> copy = new ArrayList<Text>();
			System.out.println("----------" + key + "----------");
			
			for (Text val : values) {
				System.out.println("val is " + val.toString());
				copy.add(new Text(val));
			}
			
			//ArrayList<Text> cache = new ArrayList<Text>();
			//Iterator<Text> iterator = values.iterator();
			/*
			while(iterator.hasNext()){
				copy.add(iterator.next());
			}
			*/

			System.out.println("size of values is " + copy.size());
			System.out.println("get 0: " + copy.get(0).toString() + " get 1: " + copy.get(1).toString());
			/*
			Text result = new Text(iterator.next());
			System.out.println("result is " + result.toString());
			*/
			
			
			Text result = copy.get(0);
			Text writable = copy.get(1);
			/*
			 while (iterator.hasNext()) { 
				writable = new Text(iterator.next());
				System.out.println(" first iteration: " + writable); 
				//cache.add(new Text(writable.get())); 
			 } 
			 */
			 /*
			 int size =  cache.size(); 
			 for (int i = 0; i < size; ++i) {
				System.out.println("MaxNumReducer2 second iteration: " + cache.get(i)); 
			 }
			 */
			 // result is 2,2.8932263513513514
			 //        or 1,3.43972972972973
			System.out.println("result: " + result.toString() +" writable: " + writable.toString());
			int cam_a, cam_b;
			cam_a = Integer.parseInt(result.toString().substring(0,1));
			cam_b = Integer.parseInt(writable.toString().substring(0,1));
			System.out.println("cam_a: " + cam_a +" cam_b: " + cam_b);
			double motion_a, motion_b;
			motion_a = Double.parseDouble(result.toString().substring(2));
			motion_b = Double.parseDouble(writable.toString().substring(2));
			
			/*
			 * Text motion1 = values.iterator().next(); Text motion2 =
			 * values.iterator().next().next();
			 * 
			 * if(motion1 > motion2){ context.write(key, motion1); }else {
			 * context.write(key, motion2); }
			 */
			 if(motion_a > motion_b){
				context.write(key, new IntWritable(cam_a));
			 }else{
				context.write(key, new IntWritable(cam_b));
			 }
			

		}
	}

	public int run(String[] args) throws Exception {
		Job job = new Job(getConf());
		job.setJarByClass(AutoProduce.class);
		job.setJobName("autoproduce");
		
		
		// set the priority to VERY_HIGH
		//conf.set("mapred.job.priority", JobPriority.VERY_HIGH.toString());
		//job.setPriority(JobPriority.VERY_HIGH);
		
		job.setMapOutputKeyClass(IntWritable.class);
		job.setMapOutputValueClass(Text.class);
		job.setOutputKeyClass(IntWritable.class);
		job.setOutputValueClass(IntWritable.class);

		job.setMapperClass(Map.class);
		//job.setCombinerClass(Reduce.class);
		job.setReducerClass(Reduce.class);

		job.setInputFormatClass(WholeFileInputFormat.class); // read one file each time
																// WholeFileInputFormat
																// or
																// SequenceFileInputFormat?
		job.setOutputFormatClass(TextOutputFormat.class);

		//FileInputFormat.setInputPaths(job, new Path(args[0]), new Path(args[1])); // the path
																// should be
																// /data/ap_example/cam1,
																// cam2
		FileInputFormat.addInputPaths(job, args[0]);
		Path[] inputs = FileInputFormat.getInputPaths(job);
		for (Path p : inputs) {
			System.out.println("1----file input: " + p.toString());
		}
		//FileInputFormat.addInputPath(job, new Path(args[1]));
		/*
		Path[] inputs2 = FileInputFormat.getInputPaths(job);
		for (Path p : inputs2) {
			System.out.println("2----file input: " + p.toString());
		}
		*/
		//System.out.println("input paths: " + FileInputFormat.getInputPaths(job).length());
		// or /data/ap_til/cam1, cam2
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		boolean success = job.waitForCompletion(true);
		return success ? 0 : 1;
	}

	public static void main(String[] args) throws Exception {
		int ret = ToolRunner.run(new AutoProduce(), args);
		System.exit(ret);
	}
}

class WholeFileInputFormat extends FileInputFormat<NullWritable, BytesWritable> {

	@Override
	protected boolean isSplitable(JobContext context, Path file) {
		return false;
	}

	@Override
	public RecordReader<NullWritable, BytesWritable> createRecordReader(
			InputSplit split, TaskAttemptContext context) throws IOException,
			InterruptedException {
		WholeFileRecordReader reader = new WholeFileRecordReader();
		reader.initialize(split, context);
		return reader;
	}
}

class WholeFileRecordReader extends RecordReader<NullWritable, BytesWritable> {
	private FileSplit fileSplit;
	private Configuration conf;
	private BytesWritable value = new BytesWritable();
	private boolean processed = false;

	@Override
	public void initialize(InputSplit split, TaskAttemptContext context)
			throws IOException, InterruptedException {
		this.fileSplit = (FileSplit) split;
		this.conf = context.getConfiguration();

		
		
	}

	@Override
	public boolean nextKeyValue() throws IOException, InterruptedException {
		if (!processed) {
			byte[] contents = new byte[(int) fileSplit.getLength()];
			Path file = fileSplit.getPath();
			FileSystem fs = file.getFileSystem(conf);
			FSDataInputStream in = null;
			try {
				in = fs.open(file);
				IOUtils.readFully(in, contents, 0, contents.length);
				value.set(contents, 0, contents.length);
			} finally {
				IOUtils.closeStream(in);
			}
			processed = true;
			return true;
		}
		return false;
	}

	@Override
	public NullWritable getCurrentKey() throws IOException,
			InterruptedException {
		return NullWritable.get();
	}

	@Override
	public BytesWritable getCurrentValue() throws IOException,
			InterruptedException {
		return value;
	}

	@Override
	public float getProgress() throws IOException {
		return processed ? 1.0f : 0.0f;
	}

	@Override
	public void close() throws IOException {
		// do nothing
	}
}